package com.hejie.commentservice.service.impl;

import com.hejie.commentservice.dto.CommentDTO;
import com.hejie.commentservice.entity.Comment;
import com.hejie.commentservice.repository.CommentRepository;
import com.hejie.commentservice.service.CommentService;
import com.hejie.common.exception.BusinessException;
import com.hejie.common.utils.JwtUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class CommentServiceImpl implements CommentService {
    private final CommentRepository commentRepository;
    private final RedisTemplate<String, Object> redisTemplate;
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private static final String CACHE_KEY_COMMENT = "comment:info:";
    private static final String CACHE_KEY_VIDEO_COUNT = "comment:count:video:";
    private static final long CACHE_TTL = 86400L;

    @Autowired
    public CommentServiceImpl(CommentRepository commentRepository, RedisTemplate<String, Object> redisTemplate, KafkaTemplate<String, Object> kafkaTemplate) {
        this.commentRepository = commentRepository;
        this.redisTemplate = redisTemplate;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    @Transactional
    public CommentDTO addComment(CommentDTO commentDTO) {
        // 获取当前登录用户ID
        Long userId = JwtUtils.getCurrentUserId();
        commentDTO.setUserId(userId);

        // 转换DTO为实体并保存
        Comment comment = Comment.builder()
                .videoId(commentDTO.getVideoId())
                .userId(userId)
                .parentId(commentDTO.getParentId())
                .content(commentDTO.getContent())
                .build();

        Comment savedComment = commentRepository.save(comment);
        CommentDTO result = convertToDTO(savedComment);

        // 如果是回复评论，更新父评论的回复数
        if (commentDTO.getParentId() != null) {
            commentRepository.incrementReplyCount(commentDTO.getParentId());
            // 删除父评论缓存
            redisTemplate.delete(CACHE_KEY_COMMENT + commentDTO.getParentId());
        }

        // 更新视频评论总数缓存
        redisTemplate.opsForValue().increment(CACHE_KEY_VIDEO_COUNT + commentDTO.getVideoId());

        // 缓存新评论
        redisTemplate.opsForValue().set(CACHE_KEY_COMMENT + savedComment.getId(), result, CACHE_TTL, TimeUnit.SECONDS);

        // 发布评论添加事件到Kafka
        kafkaTemplate.send("comment-added-topic", result);

        log.info("用户{}添加了评论: {}, 视频ID: {}", userId, savedComment.getId(), commentDTO.getVideoId());
        return result;
    }

    @Override
    public Page<CommentDTO> getVideoComments(Long videoId, int page, int size) {
        Pageable pageable = PageRequest.of(page, size);
        Page<Comment> commentPage = commentRepository.findByVideoIdAndParentIdIsNullOrderByCreateTimeDesc(videoId, pageable);
        return commentPage.map(this::convertToDTO);
    }

    @Override
    public Page<CommentDTO> getCommentReplies(Long parentId, int page, int size) {
        Pageable pageable = PageRequest.of(page, size);
        Page<Comment> commentPage = commentRepository.findByParentIdOrderByCreateTimeAsc(parentId, pageable);
        return commentPage.map(this::convertToDTO);
    }

    @Override
    public CommentDTO getCommentById(Long id) {
        // 先从缓存获取
        CommentDTO commentDTO = (CommentDTO) redisTemplate.opsForValue().get(CACHE_KEY_COMMENT + id);
        if (commentDTO != null) {
            return commentDTO;
        }

        // 缓存未命中，从数据库获取
        Comment comment = commentRepository.findById(id)
                .orElseThrow(() -> new BusinessException("评论不存在"));

        commentDTO = convertToDTO(comment);
        // 存入缓存
        redisTemplate.opsForValue().set(CACHE_KEY_COMMENT + id, commentDTO, CACHE_TTL, TimeUnit.SECONDS);
        return commentDTO;
    }

    @Override
    @Transactional
    public CommentDTO updateComment(Long id, CommentDTO commentDTO) {
        Long userId = JwtUtils.getCurrentUserId();
        Comment comment = commentRepository.findById(id)
                .orElseThrow(() -> new BusinessException("评论不存在"));

        // 检查权限：只有评论所有者或管理员可以修改
        if (!comment.getUserId().equals(userId) && !JwtUtils.hasAdminRole()) {
            throw new BusinessException("没有权限修改此评论");
        }

        // 更新评论内容
        comment.setContent(commentDTO.getContent());
        Comment updatedComment = commentRepository.save(comment);
        CommentDTO result = convertToDTO(updatedComment);

        // 更新缓存
        redisTemplate.opsForValue().set(CACHE_KEY_COMMENT + id, result, CACHE_TTL, TimeUnit.SECONDS);

        // 发布评论更新事件到Kafka
        kafkaTemplate.send("comment-updated-topic", result);

        log.info("用户{}更新了评论: {}", userId, id);
        return result;
    }

    @Override
    @Transactional
    public void deleteComment(Long id) {
        Long userId = JwtUtils.getCurrentUserId();
        Comment comment = commentRepository.findById(id)
                .orElseThrow(() -> new BusinessException("评论不存在"));

        // 检查权限：只有评论所有者或管理员可以删除
        if (!comment.getUserId().equals(userId) && !JwtUtils.hasAdminRole()) {
            throw new BusinessException("没有权限删除此评论");
        }

        // 逻辑删除评论
        comment.setStatus(0);
        commentRepository.save(comment);

        // 如果是回复评论，更新父评论的回复数
        if (comment.getParentId() != null) {
            commentRepository.decrementReplyCount(comment.getParentId());
            redisTemplate.delete(CACHE_KEY_COMMENT + comment.getParentId());
        }

        // 更新视频评论总数缓存
        redisTemplate.opsForValue().decrement(CACHE_KEY_VIDEO_COUNT + comment.getVideoId());

        // 删除缓存
        redisTemplate.delete(CACHE_KEY_COMMENT + id);

        // 发布评论删除事件到Kafka
        kafkaTemplate.send("comment-deleted-topic", id);

        log.info("用户{}删除了评论: {}", userId, id);
    }

    @Override
    public Page<CommentDTO> getUserComments(Long userId, int page, int size) {
        Pageable pageable = PageRequest.of(page, size);
        Page<Comment> commentPage = commentRepository.findByUserIdOrderByCreateTimeDesc(userId, pageable);
        return commentPage.map(this::convertToDTO);
    }

    @Override
    @Transactional
    public void likeComment(Long id) {
        commentRepository.incrementLikeCount(id);
        // 更新缓存
        CommentDTO commentDTO = (CommentDTO) redisTemplate.opsForValue().get(CACHE_KEY_COMMENT + id);
        if (commentDTO != null) {
            commentDTO.setLikeCount(commentDTO.getLikeCount() + 1);
            redisTemplate.opsForValue().set(CACHE_KEY_COMMENT + id, commentDTO, CACHE_TTL, TimeUnit.SECONDS);
        }
    }

    @Override
    @Transactional
    public void unlikeComment(Long id) {
        commentRepository.decrementLikeCount(id);
        // 更新缓存
        CommentDTO commentDTO = (CommentDTO) redisTemplate.opsForValue().get(CACHE_KEY_COMMENT + id);
        if (commentDTO != null) {
            commentDTO.setLikeCount(commentDTO.getLikeCount() - 1);
            redisTemplate.opsForValue().set(CACHE_KEY_COMMENT + id, commentDTO, CACHE_TTL, TimeUnit.SECONDS);
        }
    }

    @Override
    public Long countVideoComments(Long videoId) {
        // 先从缓存获取
        Object count = redisTemplate.opsForValue().get(CACHE_KEY_VIDEO_COUNT + videoId);
        if (count != null) {
            return Long.valueOf(count.toString());
        }

        // 缓存未命中，从数据库获取并缓存
        Long commentCount = commentRepository.countByVideoId(videoId);
        redisTemplate.opsForValue().set(CACHE_KEY_VIDEO_COUNT + videoId, commentCount, CACHE_TTL, TimeUnit.SECONDS);
        return commentCount;
    }

    @Override
    @Transactional
    public CommentDTO updateCommentStatus(Long id, Integer status) {
        Comment comment = commentRepository.findById(id)
                .orElseThrow(() -> new BusinessException("评论不存在"));

        comment.setStatus(status);
        Comment updatedComment = commentRepository.save(comment);
        CommentDTO result = convertToDTO(updatedComment);

        // 更新缓存
        redisTemplate.opsForValue().set(CACHE_KEY_COMMENT + id, result, CACHE_TTL, TimeUnit.SECONDS);

        log.info("管理员更新了评论{}的状态为: {}", id, status);
        return result;
    }

    // 实体转DTO
    private CommentDTO convertToDTO(Comment comment) {
        return CommentDTO.builder()
                .id(comment.getId())
                .videoId(comment.getVideoId())
                .userId(comment.getUserId())
                .parentId(comment.getParentId())
                .content(comment.getContent())
                .likeCount(comment.getLikeCount())
                .replyCount(comment.getReplyCount())
                .status(comment.getStatus())
                .createTime(comment.getCreateTime())
                .updateTime(comment.getUpdateTime())
                .build();
    }
}